package mq

import (
	log "cyiris/core/zaplog"
	"fmt"

	"github.com/streadway/amqp"
)

type Consumer struct {
	*Channel
	ExchangeName string
	ExchangeType string // "direct", "fanout", "topic" and "headers"
	RoutingKey   string
	QueueName    string
	Workers      int
	MaxPriority  int
	OnMessage    func(*amqp.Delivery, string) error //消息回调函数
}

func NewConsumer(conn *Connector, exchangeName, exchangeType, routingKey, queueName string, workers, maxPriority int, onMessage func(*amqp.Delivery, string) error) (con *Consumer, err error) {
	chann := NewChannel(conn)

	con = &Consumer{
		Channel:      chann,
		ExchangeName: exchangeName,
		ExchangeType: exchangeType,
		RoutingKey:   routingKey,
		QueueName:    queueName,
		Workers:      workers,
		MaxPriority:  maxPriority,
		OnMessage:    onMessage,
	}

	// 声明 exchange
	err = con.prepareExchange()
	if nil != err {
		return nil, err
	}

	// 声明 queue
	err = con.prepareQueue()
	if nil != err {
		return nil, err
	}

	// queue 绑定到 exchange
	err = con.bindExchangeQueue()
	if nil != err {
		return nil, err
	}

	if con.Workers <= 0 {
		con.Workers = 1
	}

	// 启动 workers 消费队列
	for i := 1; i <= con.Workers; i++ {
		go con.consume(i)
	}

	return con, nil
}

func (con *Consumer) consume(workerNum int) {
	consumerTag := con.getConsumeTag(workerNum)

	msgs, err := con.channel.Consume(con.QueueName, consumerTag, false, false, false, false, nil)
	if err != nil {
		log.Errorf("Failed to register a consumer, err:%s, queue:%v", err, con.QueueName)
		return
	}

	processing := func() {
		msg, ok := <-msgs
		if !ok {
			log.Errorf("Failed to consume msg, err:%s, queue:%v", err, con.QueueName)
			return
		}

		err = con.OnMessage(&msg, con.QueueName)

		if nil != err {
			log.Errorf("Failed to consume msg, err:%s, queue:%v", err, con.QueueName)
			return
		}

		msg.Ack(false)
	}

	for {
		processing()
	}
}

//获取 ConsumeTag
func (con *Consumer) getConsumeTag(tagId int) string {
	return fmt.Sprintf("%s_tag:%d", con.QueueName, tagId)
}

func (con *Consumer) prepareExchange() error {
	err := con.channel.ExchangeDeclare(con.ExchangeName, con.ExchangeType, true, false, false, false, nil)
	if err != nil {
		log.Errorf("Failed to declare a exchagne<%v>, err:%s", con.ExchangeName, err)
		return err
	}

	return nil
}

func (con *Consumer) prepareQueue() error {
	workerQueueOptions := map[string]interface{}{}

	if con.MaxPriority > 0 {
		workerQueueOptions["x-max-priority"] = int16(con.MaxPriority)
	}
	_, err := con.channel.QueueDeclare(con.QueueName, true, false, false, false, workerQueueOptions)
	if err != nil {
		log.Errorf("Failed to declare a queue<%v>, err:%s", con.QueueName, err)
		return err
	}

	return nil
}

func (con *Consumer) bindExchangeQueue() error {
	err := con.channel.QueueBind(con.QueueName, con.RoutingKey, con.ExchangeName, false, nil)
	if err != nil {
		log.Errorf("Failed to bind  queue<%v> to exchange<%v>, err:%s", con.QueueName, con.ExchangeName, err)
		return err
	}

	return nil
}
